package source;

import bean.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Random;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2021/11/3 9:25
 */
public class Source4_UDF {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<SensorReading> dataStream = env.addSource(new MySensorSource());


        dataStream.print("udf").setParallelism(1);

        env.execute("udf");

    }

    public static class MySensorSource implements SourceFunction<SensorReading> {

        private boolean running = true;

        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            //定义一个随机数发生器
            Random random = new Random();
            //设置10个传感器的初始值温度
            HashMap<String,Double> sensorTempMap =new HashMap<>();
            for (int i = 0; i < 10; i++) {
                sensorTempMap.put("sensor_"+(i+1),60+random.nextGaussian()*20);
            }

            while (running){
                for (String sensorId:sensorTempMap.keySet()){
                    //在当前温度基础上的随机波动
                    Double newtemp =sensorTempMap.get(sensorId)+random.nextGaussian();
                    sensorTempMap.put(sensorId,newtemp);
                    ctx.collect(new SensorReading(sensorId,System.currentTimeMillis(),newtemp));
                }
                //控制输出频率
                Thread.sleep(1000L);

             }
        }

        @Override
        public void cancel() {
            running = false;

        }
    }
}
